package com.gitee.source.practice

import java.util.UUID

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}

import scala.util.Random

/**
  * Author itcast
  * Date 2020/1/3 15:25
  * Desc 自定义数据源, 每1秒钟随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
  */
object CustomerSourcerOrderDemo {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val data: DataStream[Order] = senv.addSource(new OrderSource()).setParallelism(1)
    data.print()
    senv.execute()

  }
}

class OrderSource extends RichParallelSourceFunction[Order]() {
  var orderId: Long = 1
  var userId: String = _
  var money: Double = _
  var timestamp: Long = _
  var isRunning: Boolean = true

  override def cancel(): Unit = {
    isRunning = false
  }

  override def run(ctx: SourceFunction.SourceContext[Order]): Unit = {
    while (isRunning) {
      userId = UUID.randomUUID().toString.replaceAll("-","")
      orderId+=1
      money=Random.nextDouble()*100
      timestamp=System.currentTimeMillis()
      val order = Order(orderId,userId,money,timestamp)
      ctx.collect(order)
      Thread.sleep(1000)
    }
  }
}
//订单样例类Order(订单ID、用户ID、订单金额、时间戳)
case class Order(orderId: Long, userId: String, money: Double, timestamp: Long)